
#include <unistd.h>

#include "../iface/APRInterface.h"
#include "../iface/debug.h"
#include <apr_atomic.h>
#include <apr_file_io.h>
#include "UDPServer.h"
#include <new>

#define CONNECTION_TIMEOUT  5000
#define DATA_TIMEOUT        20000
#define POLL_TIMEOUT_USEC   5000

#define MAX_UDP_DATAGRAM_SIZE 65536

/** Constructor for the UDP server class.
 * 
 * @param inoofthreads The number of threads to use for serving
 * connections.
 * 
 * @param iportno The port to listen on.
 * 
 * @param ihostname The host (interface) to listen on.
 * 
 * @param upperpool The memory pool to derive this server memory pool
 * from. */
UDPServer::UDPServer(int inoofthreads, int iportno, const char *ihostname, apr_pool_t *upperpool)
{
   apr_status_t status;
   status = apr_pool_create(&mp, upperpool);
   noofthreads = inoofthreads;
   listenstop = 0;
   sessionstop = 0;
   portno = iportno;
   listenstop = 1;
   if (ihostname)
   {
      hostname = (char *)apr_palloc(mp, sizeof(char)*(strlen(ihostname)+1));
      strcpy(hostname, ihostname);
   } else hostname = NULL;
   pipearray = (ClientPipeDatagrams *)apr_palloc(mp, sizeof(ClientPipeDatagrams)*noofthreads);
   for (int i = 0; i < noofthreads; i++)
   {
      pipearray[i].datagram_data = NULL;
      pipearray[i].datagram_length = 0;
      pipearray[i].complete = 1;
      void *mem_for_new_sessionstart_lock = apr_palloc(mp, sizeof(APRMutex));
      pipearray[i].sessionstart = new(mem_for_new_sessionstart_lock) APRMutex(mp);
      pipearray[i].localmp = NULL;
   }
   session_threads = (apr_thread_t **)apr_palloc(mp, sizeof(apr_thread_t *)*noofthreads);
   session_threadattrs = (apr_threadattr_t **)apr_palloc(mp, sizeof(apr_threadattr_t *)*noofthreads);
}

/** Destructor for the basic UDP server. */
UDPServer::~UDPServer()
{
   joinThreads();
   // don't do free on pipearray, session_threads
   apr_pool_destroy(mp);
}

/** Start the listening thread. This must be started first. */
int UDPServer::startListeningThread()
{
   apr_status_t status;
   status = apr_threadattr_create(&listening_attr, mp);
   CheckAPRError(status);
   status = apr_thread_create(&listening_thread, listening_attr, (functor)(&(UDPServer::handleServerListening)), (void *)this, mp);
   CheckAPRError(status);
   return 0;
}

/** Start the session threads. These must be started after the
 * listening thread. */
int UDPServer::startSessionThreads()
{
   void ***tempvoid;
   tempvoid = (void ***)apr_palloc(mp, sizeof(void **)*noofthreads);
   apr_status_t status;
   for (int i = 0; i < noofthreads; i++)
   {
      tempvoid[i] = (void **)apr_palloc(mp, sizeof(void *)*2);
      tempvoid[i][0] = (void *)apr_palloc(mp, sizeof(void *));
      tempvoid[i][1] = (void *)apr_palloc(mp, sizeof(int));
      *((void **)tempvoid[i][0]) = (void *)this;
      *((int *)tempvoid[i][1]) = i;
      status = apr_threadattr_create(&(session_threadattrs[i]), mp);
      CheckAPRError(status);
      // Create stack protectors to prevent stack smashing attacks
      //status = apr_threadattr_guardsize_set(session_threadattrs[i], 4096);
      //CheckAPRError(status);
      status = apr_thread_create(&(session_threads[i]), session_threadattrs[i], (functor)(getClientHandlerFunction()), (void *)tempvoid[i], mp);
      CheckAPRError(status);
   }
   return 0;
}

/** Accept the incoming connection on the socket.
 * 
 * @param pollset Pointer to the set of existing connections to be
 * polled for data.
 * 
 * @param lsock The socket pointer.
 * 
 * @param mp Memory pool from which to allocate new client handler
 * pipes. */
void UDPServer::acceptIncomingOnSocket(apr_pollset_t *pollset, apr_socket_t *lsock, apr_pool_t *mp)
{
   // Must find inactive socket, or this does nothing (which is good!)
   apr_status_t status;
   for (int j = 0; j < noofthreads; j++)
   {
      if (pipearray[j].complete)
      {
         pipearray[j].sessionstart->lock();
         if (pipearray[j].localmp != NULL)
         {
            if (pipearray[j].datagram_data) free(pipearray[j].datagram_data);
            pipearray[j].datagram_length = 0;
            apr_pool_clear(pipearray[j].localmp);
            apr_pool_destroy(pipearray[j].localmp);
         }
         status = apr_pool_create(&(pipearray[j].localmp), mp);
         CheckAPRError(status);
         char *datagram_data_2 = (char *)malloc(sizeof(char)*MAX_UDP_DATAGRAM_SIZE);
         apr_size_t dataquant = MAX_UDP_DATAGRAM_SIZE;
         pipearray[j].clisock = lsock;
         status = apr_socket_recvfrom(&(pipearray[j].cliaddr), pipearray[j].clisock, 0, datagram_data_2, &dataquant);
         pipearray[j].datagram_data = (char *)malloc(sizeof(char)*(dataquant+1));
         memcpy(pipearray[j].datagram_data, datagram_data_2, sizeof(char)*dataquant);
         free(datagram_data_2);
         pipearray[j].datagram_data[dataquant] = '\0';
         pipearray[j].datagram_length = dataquant;
         pipearray[j].complete = 0;
         pipearray[j].sessionstart->signal();
         pipearray[j].sessionstart->unlock();
         return;
      }
   }
}

/** Start listening on the socket (the listening thread is this
 * function).
 * 
 * @param tthread The APR pointer to this thread grabbed.
 * 
 * @param vthis The this pointer carried into the thread.
 * 
 * @return Return is NULL. */
void *UDPServer::handleServerListening(apr_thread_t *tthread, void *vthis)
{
   apr_status_t status;
   UDPServer *kthis = (UDPServer *)vthis;
   apr_pool_t *upperpool, *mp;
   upperpool = apr_thread_pool_get(tthread);
   apr_pool_create(&mp, upperpool);

   apr_socket_t *lsock = kthis->createListeningSocket(mp);
   if (lsock == NULL)
   {
      kthis->listenstop = 1;
      return NULL;
   }
   apr_pollset_t *pollset;

   status = apr_pollset_create(&pollset, kthis->noofthreads, mp, 0);
   CheckAPRError(status);
   apr_pollfd_t mainpollfd = { mp, APR_POLL_SOCKET, APR_POLLIN, 0, { NULL }, NULL };
   const apr_pollfd_t *workingpollfd;
   mainpollfd.desc.s = lsock;
   status = apr_pollset_add(pollset, &mainpollfd);
   CheckAPRError(status);

   while (!(kthis->listenstop))
   {
      apr_int32_t fdnum;
      // Using same timeout for polling as for connections!
      status = apr_pollset_poll(pollset, POLL_TIMEOUT_USEC, &fdnum, &workingpollfd);
      if (!(APR_STATUS_IS_EINTR(status) || APR_STATUS_IS_TIMEUP(status)))
         CheckAPRError(status);
      for (int i = 0; i < fdnum; i++)
      {
         int matched = 0;
         if (workingpollfd[i].desc.s == lsock)
         {
            kthis->acceptIncomingOnSocket(pollset, lsock, mp);
            matched = 1;
         }
         if (matched == 0)
            errprint("Unmatched/unknown file descriptor in poll.");
      }
   }
   kthis->listenstop = 1;
   return NULL;
}

/** Start handling clients to the server (the client handler threads
 * run this).
 * 
 * @param threadid The integer thread ID number.
 * 
 * @param localmp The memory pool local to this thread. */
void UDPServer::handleClientsOnThread(int threadid, apr_pool_t *localmp)
{
   pipearray[threadid].sessionstart->lock();
   while (!sessionstop)
   {
      pipearray[threadid].sessionstart->wait();
      if (sessionstop) break;
      apr_pool_t *sp;
      apr_pool_create(&sp, localmp);
      int has_read = 0, has_written = 0, has_finalised = 0;
      if (pipearray[threadid].datagram_length > 0)
      {
         printf("%s\n", pipearray[threadid].datagram_data);
      }
      pipearray[threadid].complete = 1;
      apr_pool_clear(sp);
      apr_pool_destroy(sp);
   }
   pipearray[threadid].sessionstart->unlock();
}

/** Start handling clients to the server (the client handler threads
 * run this).
 * 
 * @param tthread The APR pointer to this thread grabbed.
 * 
 * @param args The void pointer given to the thread. This is actually
 * a pointer to an array of two void pointers, which are the this
 * pointer and a pointer to the thread ID respectively.
 * 
 * @return Return is NULL. */
void *UDPServer::handleClientsOnThreadStatic(apr_thread_t *tthread, void *args)
{
   apr_status_t status;
   UDPServer *kthis = *((UDPServer **)(((void **)(args))[0]));
   int i = *((int *)(((void **)(args))[1]));
   apr_pool_t *upperpool, *mp;
   upperpool = apr_thread_pool_get(tthread);
   status = apr_pool_create(&mp, upperpool);
   CheckAPRError(status);
   
   kthis->handleClientsOnThread(i, mp);

   apr_pool_clear(mp);
   apr_pool_destroy(mp);
   ERR_remove_state(0);
   apr_thread_exit(tthread, APR_SUCCESS);
   return NULL;
}

/** Join threads to main thread (stop all sessions). */
int UDPServer::joinThreads()
{
   sessionstop = 1;
   apr_status_t status;
   for (int i = 0; i < noofthreads; i++)
   {
      if (pipearray[i].sessionstart)
         pipearray[i].sessionstart->signal();
      apr_thread_join(&status, session_threads[i]);
      CheckAPRError(status);
   }

   listenstop = 1;
   apr_thread_join(&status, listening_thread);
   CheckAPRError(status);
   for (int i = 0; i < noofthreads; i++)
   {
      if (pipearray[i].localmp != NULL)
      {
         if (pipearray[i].datagram_data) free(pipearray[i].datagram_data);
         pipearray[i].datagram_data = NULL;
         pipearray[i].datagram_length = 0;
         apr_pool_clear(pipearray[i].localmp);
         apr_pool_destroy(pipearray[i].localmp);
      }
   }
   return 0;
}

/** Make a listening socket from the memory pool
 * 
 * @param mp Memory pool to use.
 * 
 * @return Pointer to new listening socket. */
apr_socket_t *UDPServer::createListeningSocket(apr_pool_t *mp)
{
   apr_status_t status;
   apr_socket_t *tsocket;
   apr_sockaddr_t *addrstruct;

   // NULL here gives the 0.0.0.0 that is binding to all network
   // interfaces
   status = apr_sockaddr_info_get(&addrstruct, hostname, APR_INET, portno, 0, mp);
   CheckAPRError(status);
   status = apr_socket_create(&tsocket, addrstruct->family, SOCK_DGRAM, APR_PROTO_UDP, mp);
   CheckAPRError(status);
   status = apr_socket_opt_set(tsocket, APR_SO_NONBLOCK, 1);
   CheckAPRError(status);
   status = apr_socket_timeout_set(tsocket, CONNECTION_TIMEOUT);
   CheckAPRError(status);
   status = apr_socket_opt_set(tsocket, APR_SO_REUSEADDR, 1);
   CheckAPRError(status);
   status = apr_socket_bind(tsocket, addrstruct);
   CheckAPRError(status);
   if (status != APR_SUCCESS)
   {
      errprint("bind to '%s:%d' failed", hostname, portno);
      return NULL;
   }

   listenstop = 0;
   return tsocket;
}

/** Get the static client address string from the client address
 * pointer in the pipearray. The returned pointer is only valid for as
 * long as the socket remains open.
 * 
 * @param threadid The ID of the thread who client address is to be
 * obtained.
 * 
 * @return Pointer to the temporarily available string. */
const char *UDPServer::getClientAddressAsStaticString(int threadid)
{
   apr_status_t status;
   char *ipaddress;
   status = apr_sockaddr_ip_get(&ipaddress, &(pipearray[threadid].cliaddr));
   CheckAPRError(status);
   return ipaddress;
}
